package org.example.lookup_join;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkUserTableInit {
    public static void main(String[] args) {
        // 设置批处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.BATCH); // 设置为批处理模式
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inBatchMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 创建Paimon目录
        String createCatalogSQL = "CREATE CATALOG paimon WITH (" +
                "    'type' = 'paimon'," +
                "    'warehouse' = 'file:///tmp/paimon'" +
                ");";
        System.out.println("Executing SQL: " + createCatalogSQL);
        tableEnv.executeSql(createCatalogSQL);

        // 使用Paimon目录
        String useCatalogSQL = "USE CATALOG paimon;";
        System.out.println("Executing SQL: " + useCatalogSQL);
        tableEnv.executeSql(useCatalogSQL);

        // 删除已存在的用户表
        String dropTableSQL = "DROP TABLE IF EXISTS user_table;";
        System.out.println("Executing SQL: " + dropTableSQL);
        tableEnv.executeSql(dropTableSQL);

        // 创建Paimon用户表（必须满足LOOKUP JOIN要求）
        String createUserTableSQL = "CREATE TABLE IF NOT EXISTS `paimon`.`default`.`user_table` (" +
                "    user_id BIGINT PRIMARY KEY NOT ENFORCED, " + // 必须显式定义主键
                "    user_name STRING, " + // 用户名
                "    user_email STRING, " + // 用户邮箱
                "    user_address STRING " + // 用户地址
                ") WITH (" +
                "    'bucket' = '-1'" + 
                ");";
        System.out.println("Executing SQL: " + createUserTableSQL);
        tableEnv.executeSql(createUserTableSQL);

        // 构建批量插入SQL语句
        StringBuilder insertDataSQL = new StringBuilder("INSERT INTO user_table VALUES ");
        for (int i = 1; i <= 100; i++) {
            String userAddress = "Address_" + (int)(Math.random() * 100 + 1) + ", City_" + (int)(Math.random() * 10 + 1);
            insertDataSQL.append("(")
                    .append(i).append(", ") // user_id
                    .append("'user_").append(i).append("', ") // user_name
                    .append("'user").append(i).append("@example.com', ") // user_email
                    .append("'").append(userAddress).append("')"); // user_address
            if (i < 100) {
                insertDataSQL.append(", ");
            }
        }
        System.out.println("Executing SQL: " + insertDataSQL);
        tableEnv.executeSql(insertDataSQL.toString());

        // 测试查询
        String testQuerySQL = "SELECT * FROM user_table LIMIT 10;";
        System.out.println("Executing SQL: " + testQuerySQL);
        tableEnv.executeSql(testQuerySQL).print();
    }
}
